package org.softee.functional;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;

import org.softee.functional.exception.ParallelException;
import org.softee.functional.functions.Mapper;
import org.softee.functional.functions.Predicate;
import org.softee.functional.functions.ReduceFunction;
/**
 * Parallel strategy: Atomize the tasks into individual payloads and perform computation on each using executor.
 *
 * @author morten.hattesen@gmail.com
 *
 */
public class ParallelAtomized<F> extends AbstractParallelFunctional<F> {
    private static final int REDUCE_RECURSE_SIZE_LIMIT = 4;

    public ParallelAtomized(Collection<F> data) {
        super(data);
    }

    public ParallelAtomized(Collection<F> data, ExecutionContext executionContext) {
        super(data, executionContext);
    }

    @Override
    public ParallelFunctional<F> filter(final Predicate<? super F> predicate) {
        if (data.size() <= 1) {
            return withData(doFilter(predicate, data));
        }

        // TODO optimize to be single-step process
        Executor executor = executionContext.getExecutor();

        // Add all the future results to a separate collection
        List<Future<Boolean>> futureResults = new ArrayList<Future<Boolean>>(data.size());
        for (final F input: data) {
            Callable<Boolean> callable = new Callable<Boolean>() {
                @Override
                public Boolean call() throws Exception {
                    return predicate.apply(input);
                }
            };
            FutureTask<Boolean> futureTask = new FutureTask<Boolean>(callable);
            futureResults.add(futureTask);
            executor.execute(futureTask);
        }

        // Collect all future results
        Collection<F> results = new ArrayList<F>(data.size());
        Iterator<F> inputIterator = data.iterator();
        for (Future<Boolean> future : futureResults) {
            F value = inputIterator.next();
            try {
                if (future.get()) {
                    results.add(value);
                }
            } catch (InterruptedException e) {
                // propagate
                Thread.currentThread().interrupt();
                abort(e, futureResults);
            } catch (ExecutionException e) {
                // TODO abort pending futures
                abort(e.getCause(), futureResults);
            }
        }

        return withData(results);
    }

    @Override
    public <T> ParallelFunctional<T> map(final Mapper<? super F, ? extends T> mapper) {
        if (data.size() <= 1) {
            return withData(doMap(mapper, data));
        }

        // Add all the future results to a separate collection
        Collection<Future<T>> futureResults = new ArrayList<Future<T>>(data.size());
        for (final F input: data) {
            Callable<T> callable = new Callable<T>() {
                @Override
                public T call() throws Exception {
                    return mapper.apply(input);
                }
            };
            FutureTask<T> futureTask = new FutureTask<T>(callable);
            futureResults.add(futureTask);
            executionContext.getExecutor().execute(futureTask);
        }

        // Collect all future results
        ArrayList<T> results = new ArrayList<T>(data.size());
        for (Future<T> futureResult : futureResults) {
            T result;
            try {
                result = futureResult.get();
                results.add(result);
            } catch (InterruptedException e) {
                // propagate
                Thread.currentThread().interrupt();
                abort(e, futureResults);
            } catch (ExecutionException e) {
                abort(e.getCause(), futureResults);
            }
        }

        return withData(results);
    }

    @Override
    public F reduce(ReduceFunction<F> reduceFunction) {
        if (data.size() < REDUCE_RECURSE_SIZE_LIMIT) {
            return doReduce(reduceFunction, data);
        }

        List<F> inputList = (data instanceof List) ? (List<F>)data : new ArrayList<F>(data);
        try {
            Future<F> futureResult = reduceRecursive(executionContext.getExecutor(), inputList, reduceFunction);
            F result = futureResult.get();
            //System.out.println("Result " + result);
            return result;
        } catch (InterruptedException e) {
            // propagate
            Thread.currentThread().interrupt();
            throw new ParallelException(e);
        } catch (ExecutionException e) {
            // TODO abort pending futures
            throw new ParallelException(e.getCause());
        }
    }

    Future<F> reduceRecursive(Executor executor, List<F> inputList, ReduceFunction<F> reduceFunction)
            throws InterruptedException, ExecutionException {
        //System.out.println("Reducing " + inputList);
        if (inputList.size() < REDUCE_RECURSE_SIZE_LIMIT) {
            FutureTask<F> futureTask = reduceFuture(reduceFunction, inputList);
            executor.execute(futureTask);
            return futureTask;
        }
        // divide and recurse
        int index = inputList.size() / 2;
        Future<F> f1 = reduceRecursive(executor, inputList.subList(0, index), reduceFunction);
        Future<F> f2 = reduceRecursive(executor, inputList.subList(index, inputList.size()), reduceFunction);

        // reduce the resulting pair of values
        @SuppressWarnings("unchecked")
        List<F> pair = (List<F>) Arrays.asList(new Object[] {f1.get(), f2.get()});
        //System.out.println("Merging " + pair);
        FutureTask<F> futureTask = reduceFuture(reduceFunction, pair);
        executor.execute(futureTask);
        return futureTask;
    }

    private FutureTask<F> reduceFuture(final ReduceFunction<F> reduceFunction, final Collection<F> data) {
        Callable<F> callable = new Callable<F>() {
            @Override
            public F call() throws Exception {
                return reduceFunction.apply(data);
            }
        };
        return new FutureTask<F>(callable);
    }

    @Override
    protected <T> ParallelFunctional<T> withData(Collection<T> data) {
        return new ParallelAtomized<T>(data, executionContext);
    }

    @Override
    protected ParallelFunctional<F> withExecutionContext(ExecutionContext executionContext) {
        return new ParallelAtomized<F>(data, executionContext);
    }
}
